package org.example.consumer;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

@Slf4j
public class SimpleConsumer {
    @SneakyThrows
    public static void createConsumer() {
        // 创建消费组
        DefaultMQPushConsumer niweiConsumerGroup = new DefaultMQPushConsumer("niwei_consumer_group");
        niweiConsumerGroup.setNamesrvAddr("192.168.31.104:9876");
        // 设置消费顺序
        niweiConsumerGroup.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        // 订阅topic下所有数据
        niweiConsumerGroup.subscribe("topic", "*");

        niweiConsumerGroup.registerMessageListener((List<MessageExt> list, ConsumeConcurrentlyContext context) -> {
            if (list != null) {
                for (MessageExt messageExt : list) {
                    System.out.println(new String(messageExt.getBody()));

                }
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });

        niweiConsumerGroup.start();

    }
}
